import pyspark
from pyspark import SparkContext, RDD
from pyspark import SparkConf
from pyspark.sql import SparkSession
from pyspark.streaming import StreamingContext
import pandas as pd
import pymongo


def split0(x):
    data = x.split(",")
    data[6] = int(data[6])
    data[7] = int(data[7])
    return data


def educationCout(x):
    num = x[1][0]
    avgSalary = x[1][1] / x[1][0]
    myclient = pymongo.MongoClient("mongodb://39.99.155.45:27017")
    mydb = myclient["Result"]
    mycol = mydb["educationResult"]
    mycol.delete_one({"education": x[0]})
    mycol.insert_one({"education": x[0], "num": num, "avgSalary": avgSalary})


def aboutEducation(rdd: RDD):
    num = rdd.count()
    if num > 0:
        data = rdd.map(lambda x: (x[3], x[6]))
        num1 = data.reduceByKey(lambda x, y: x + y)
        data = rdd.map(lambda x: (x[3], x[6] * x[7]))
        salary = data.reduceByKey(lambda x, y: x + y)
        result = num1.join(salary)
        result.foreach(educationCout)


def aboutCity(rdd: RDD):
    num = rdd.count()
    if num > 0:
        data = rdd.map(lambda x: (x[1], x[6]))
        num1 = data.reduceByKey(lambda x, y: x + y)
        data = rdd.map(lambda x: (x[1], x[6] * x[7]))
        salary = data.reduceByKey(lambda x, y: x + y)
        result = num1.join(salary)
        result.foreach(cityCout)


def cityCout(x):
    num = x[1][0]
    avgSalary = x[1][1] / x[1][0]
    myclient = pymongo.MongoClient("mongodb://39.99.155.45:27017")
    mydb = myclient["mytest"]
    mycol = mydb["cityResult"]
    mycol.delete_one({"city": x[0]})
    mycol.insert_one({"city": x[0], "num": num, "avgSalary": avgSalary})


if __name__ == '__main__':
    sparkConf = SparkConf().setMaster("spark://192.168.43.147:7077").setAppName("test1")
    sc = SparkContext(conf=sparkConf)
    ssc = StreamingContext(sc, 5)
    dirStream = ssc.textFileStream(".\data")
    stream = dirStream.map(split0)
    stream.foreachRDD(aboutEducation)
    stream.foreachRDD(aboutCity)
    stream.pprint()

    # cityData.pprint()

    # for i in range(0, len(citys)):
    #     cityData = dirStream.filter(lambda x: citys[i] in x)
    ssc.start()
    ssc.awaitTermination()
